package com.helei.controller;

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * @date 2024/10/2 16:42
 */
@RestController
@RequestMapping("/rocketmq")
public class ProducerRocketMq {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送同步消息
     */
    @GetMapping("/sendSynMessage")
    public void sendSynMessage() {
        SendResult sendResult = rocketMQTemplate.syncSend("sync-topic-boot", "这是boot结合的一个消息");
        System.out.println("消息msgId" + sendResult.getMsgId());
        System.out.println("消息的状态---->" + sendResult.getSendStatus());
    }

    /**
     * 发送异步消息
     */
    @GetMapping("/sendAsynMessage")
    public void sendAsynMessage() {
        rocketMQTemplate.asyncSend("asyn-topic-boot", "异步消息", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("异步消息发送成功---" + sendResult.getSendStatus());
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("异步消息发送失败");
                System.out.println(throwable.getMessage());
            }
        });
    }

    /**
     * 发送单向消息
     */
    @GetMapping("/sendOneWayMessage")
    public void sendOneWayMessage() {
        rocketMQTemplate.sendOneWay("oneway-topic-boot", "单向消息");
    }

    /**
     * 发送延迟消息
     * 设置延迟等级  messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    @GetMapping("/sendDelayMessage")
    public void sendDelayMessage() {
        //创建消息
        LocalDateTime now = LocalDateTime.now();
        String format = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").format(now);
        Message<String> message = MessageBuilder.withPayload("延迟消息:"+format).build();
        SendResult sendResult = rocketMQTemplate.syncSend("delay-topic-boot", message, 50000, 5);
        System.out.println("发送成功id: "+sendResult.getMsgId()+",,,"+format);
    }


    /**
     * key值发送消息
     */
    @GetMapping("/sendKeyMessage")
    public void sendKeyMessage() {

        //主题上打key值
        Message<String> message = MessageBuilder.withPayload("这是一个携带key的消息")
                .setHeader(RocketMQHeaders.KEYS, "keyValue").build();

        // 主题上打标签发送消息keyA
        rocketMQTemplate.syncSend("key-topic-boot", message);
    }

    /**
     * 重试消费
     */

    @GetMapping("/retryMessage")
    public void retryMessage() {
        Message<String> message = MessageBuilder.withPayload("重试消息").build();
        rocketMQTemplate.syncSend("retry-topic-boot", message);
    }
    /**
     * 定时发送消息
     */

    @GetMapping("/deliverTimeMessage")
    public void deliverTimeMessage() {
        Message<String> message = MessageBuilder.withPayload("定时发送消息").build();
        // 延迟15s发送消息
        SendResult sendResult = rocketMQTemplate.syncSendDeliverTimeMills("deliver-topic-boot", message, System.currentTimeMillis() + 1000 * 15);
        System.out.println(sendResult.getSendStatus());
    }
}

